/*
 * Copyright (C) 2017 Honeywell, Inc. All Rights Reserved.
 */

package org.thanos.demo.nats;

import io.nats.client.Message;
import io.nats.client.MessageHandler;

public abstract class NatsConsumerBase {

    private NatsClient natsClient;
    private String topic;
    private String queue;
    private Boolean async;
    private AsyncNatsConsumerPool asyncNatsConsumerPool;

    public NatsConsumerBase() {
        async = false;
    }

    public void init() throws InterruptedException {
        MessageHandler messageHandler;
        if (async) {
            messageHandler = msg -> asyncNatsConsumerPool.submit(msg, this::consume);
        } else {
            messageHandler = this::consume;
        }
        natsClient.subscribe(topic, queue, messageHandler);
    }

    protected abstract void consume(Message message);

    public NatsClient getNatsClient() {
        return natsClient;
    }

    public void setNatsClient(NatsClient natsClient) {
        this.natsClient = natsClient;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getQueue() {
        return queue;
    }

    public void setQueue(String queue) {
        this.queue = queue;
    }

    public Boolean getAsync() {
        return async;
    }

    public void setAsync(Boolean async) {
        this.async = async;
    }

    public AsyncNatsConsumerPool getAsyncNatsConsumerPool() {
        return asyncNatsConsumerPool;
    }

    public void setAsyncNatsConsumerPool(AsyncNatsConsumerPool asyncNatsConsumerPool) {
        this.asyncNatsConsumerPool = asyncNatsConsumerPool;
    }
}
